package hyl.ext.pipe.demo;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import hyl.core.MyFun;
import hyl.core.info.CMsg;

public class testdemo1 {
	//private static final Logger log = LoggerFactory.getLogger(testdemo1.class);
	public static volatile long _ms = MyFun.getMs();
	
	public void fun1() {
		// MyDate.sleep(1000);
		System.out.println("fun1");
	}

	public void fun2(CMsg msg) {
		// MyDate.sleep(1000);
		msg._m属性.put("第2步", "ok");
		msg.bs内容 = (new String(msg.bs内容) + ">2").getBytes();
		/// System.out.println("fun2::"+msg.toString());
	}

	public void fun3(CMsg msg) {
		// MyDate.sleep(1000);
		msg.bs内容 = (new String(msg.bs内容) + ">3").getBytes();
		msg._m属性.put("第3步", "ok");
		// System.out.println("fun3::"+msg.toString());
	}

	public void fun4(CMsg msg) {
		// MyDate.sleep(1000);
		msg.bs内容 = (new String(msg.bs内容) + ">4").getBytes();
		msg._m属性.put("第4步", "ok");
		System.out.println("fun4::"+msg);
	}

	// 每批10条处理 用例
	AtomicInteger _atomicI5 = new AtomicInteger(0);
	List<CMsg> msg5 =  Collections.synchronizedList(new ArrayList<>());
	public synchronized void fun5(CMsg msg) {

		// MyDate.sleep(1000);
		msg.bs内容 = (new String(msg.bs内容) + ">5").getBytes();
		msg._m属性.put("第5步", "ok");
		msg5.add(msg);
		int k=_atomicI5.incrementAndGet() ;
		
		if (k%100==0) {
			//log.info("批量处理5::" +k);
			msg5.clear();

		}
		// System.out.println("fun5::"+msg.toString());
	}

	// 每隔3s 或者 超过10条 批量处理用例
	AtomicInteger _atomicI6 = new AtomicInteger(0);
	List<CMsg> msg6 =  Collections.synchronizedList(new ArrayList<>());

	public synchronized void fun6(CMsg msg) {
		// MyDate.sleep(1000);

		msg.bs内容 = msg.bs内容;
		msg._m属性.put("第6步", "ok");
		//System.out.println("fun6::" + _atomicI6.get() + " " + msg.toString());
		msg6.add(msg);
		long ms=MyFun.getMs() ;
		if (ms> _ms + 100 || _atomicI6.incrementAndGet() %10==0) {
			//log.info("批量处理6::" + JSON.toJSONString(msg6));
			_ms=ms;
			msg6.clear();

		}
	}
	

}
